package com.cctc.rds.scheduler.client.raft;

import com.cctc.rds.nrpc.api.client.config.RpcClientConfig;
import com.cctc.rds.nrpc.api.exception.ConnectionTimeoutException;
import com.cctc.rds.nrpc.api.exception.RpcIOException;
import com.cctc.rds.nrpc.api.rpc.RpcRequest;
import com.cctc.rds.nrpc.api.rpc.RpcResponse;
import com.cctc.rds.nrpc.core.client.NettyRpcClient;
import com.cctc.rds.nrpc.core.util.JsonUtils;

import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 任务实例日志 RPC [请求] 客户端。
 */
public class NettyRpcClientHolder {
    private NettyRpcClient[] clients;
    private RpcClientConfig[] configs;
    private AtomicReference<NettyRpcClient> leader = new AtomicReference<>(null);

    public NettyRpcClientHolder(RpcClientConfig[] configs) {
        this.configs = configs;
        this.clients = new NettyRpcClient[this.configs.length];
        for (int i = 0; i < clients.length; i++) {
            this.clients[i] = new NettyRpcClient();
        }
    }

    public void start() {
        for (int i = 0; i < clients.length; i++) {
            try {
                clients[i].start(configs[i]);
                waitReady(clients[i], 15000);
            } catch (Exception ignored) {
            }
        }
    }

    public RpcResponse removeInvoke(RpcRequest rpcRequest) throws RpcIOException {
        NettyRpcClient client = getReadyClient();
        try {
            RpcResponse rpcResponse = client.remoteInvoke(rpcRequest);
            if (rpcResponse.getException() != null) {
                if (redirectByException(rpcResponse.getException(), client)) {
                    return leader.get().remoteInvoke(rpcRequest);
                }
            }
            return rpcResponse;
        } catch (RpcIOException e) {
            if (e instanceof ConnectionTimeoutException) {
                leader.compareAndSet(client, null);
            }
            throw e;
        }
    }

    public void stop() {
        for (NettyRpcClient client : clients) {
            try {
                client.stop();
            } catch (Exception ignored) {
            }
        }
    }

    private NettyRpcClient getReadyClient() throws RpcIOException {
        if (leader.get() != null) return leader.get();
        synchronized (this) {
            int waitTimeCount = 0;
            while (waitTimeCount < configs[0].getConnectionTimeoutMs()) {
                for (NettyRpcClient client : clients) {
                    if (client.isReady()) return client;
                }
                try {
                    Thread.sleep(200);
                    waitTimeCount += 200;
                } catch (InterruptedException e) {
                }
            }
        }
        throw new ConnectionTimeoutException("无可用连接...");
    }

    private boolean redirectByException(Throwable e, NettyRpcClient client) {
        try {
            Throwable caused = e;
            while (caused != null) {
                Map<String, Object> resource = JsonUtils.fromJson(e.getMessage(), Map.class);
                if (resource != null && resource.containsKey("redirectToLeader")) {
                    Map<String, Object> redirectToLeader = (Map<String, Object>) resource.get("redirectToLeader");
                    String ip = (String) redirectToLeader.get("ip");
                    int port = Integer.parseInt(redirectToLeader.get("port").toString());
                    resetLeader(ip, port);
                    return leader.get() != client;
                }
            }
        } catch (Throwable ignored) {
        }
        return false;
    }

    private synchronized void resetLeader(String ip, int port) {
        NettyRpcClient curLeader = leader.get();
        for (int i = 0; i < configs.length; i++) {
            RpcClientConfig config = configs[i];
            if (config.getRemoteIp().equals(ip) && config.getRemotePort() == port) {
                System.out.println("redirect to leader -> " + ip + ":" + port);
                leader.compareAndSet(curLeader, clients[i]);
            }
        }
    }

    private static void waitReady(NettyRpcClient client, long timeout) {
        long count = 0;
        while (!client.isReady() && count < timeout) {
            try {
                Thread.sleep(100);
                count += 100;
            } catch (InterruptedException ignored) {
            }
        }
    }

}
